src/handlers/http/ingest.rs (1)

256-270: OTEL ingestion should also pass effective time_partition

To uniformly enforce time-partition validation for OTEL, compute the same effective time-partition (header override or stream default) and pass it to flatten_and_push_logs.

-                flatten_and_push_logs(
-                    serde_json::from_slice(&body)?,
-                    stream_name,
-                    log_source,
-                    &p_custom_fields,
-                    None,
-                )
+                let time_partition = req
+                    .headers()
+                    .get("x-p-time-partition")
+                    .and_then(|h| h.to_str().ok())
+                    .map(|s| s.to_string())
+                    .or_else(|| PARSEABLE.get_stream(stream_name).ok()?.get_time_partition());
+
+                flatten_and_push_logs(
+                    serde_json::from_slice(&body)?,
+                    stream_name,
+                    log_source,
+                    &p_custom_fields,
+                    time_partition,
+                )
                 .await?;

src/handlers/http/modal/utils/ingest_utils.rs (1)

133-139: Regression: stream-configured time_partition no longer enforced when caller passes None

push_logs now only uses the passed time_partition and no longer derives it from the stream. If call sites forget to pass it (several currently pass None), time-partition validation silently won’t run. To preserve correctness:

• Compute an effective time partition: per-request override if present, else stream-configured.
• Use that for both convert_array_to_object and into_event.

 pub async fn push_logs(
     stream_name: &str,
     json: Value,
     log_source: &LogSource,
     p_custom_fields: &HashMap<String, String>,
-    time_partition: Option<String>,
+    time_partition: Option<String>,
 ) -> Result<(), PostError> {
     let stream = PARSEABLE.get_stream(stream_name)?;
     let time_partition_limit = PARSEABLE
         .get_stream(stream_name)?
         .get_time_partition_limit();
     let static_schema_flag = stream.get_static_schema_flag();
     let custom_partition = stream.get_custom_partition();
     let schema_version = stream.get_schema_version();
     let p_timestamp = Utc::now();
 
-    let data = convert_array_to_object(
+    // Prefer per-request override, else fall back to the stream-configured time partition
+    let effective_time_partition = time_partition.or_else(|| stream.get_time_partition());
+
+    let data = convert_array_to_object(
         json,
-        time_partition.as_ref(),
+        effective_time_partition.as_ref(),
         time_partition_limit,
         custom_partition.as_ref(),
         schema_version,
         log_source,
     )?;
 
     for json in data {
         let origin_size = serde_json::to_vec(&json).unwrap().len() as u64; // string length need not be the same as byte length
         let schema = PARSEABLE.get_stream(stream_name)?.get_schema_raw();
         json::Event { json, p_timestamp }
             .into_event(
                 stream_name.to_owned(),
                 origin_size,
                 &schema,
                 static_schema_flag,
                 custom_partition.as_ref(),
-                time_partition.as_ref(),
+                effective_time_partition.as_ref(),
                 schema_version,
                 StreamType::UserDefined,
                 p_custom_fields,
             )?
             .process()?;
     }
     Ok(())
 }

Also applies to: 149-156, 168-169

src/catalog/mod.rs (1)

90-109: Unsafe unwraps in get_file_bounds can crash the node

find(...).unwrap() and stats.as_ref().unwrap() will panic if:

• the expected column is missing in a file, or
• stats are absent (e.g., file with schema/metadata drift).

Prefer returning a handled error or skipping such files with a log. Minimal fix:

-fn get_file_bounds(
-    file: &manifest::File,
-    partition_column: String,
-) -> (DateTime<Utc>, DateTime<Utc>) {
-    match file
-        .columns()
-        .iter()
-        .find(|col| col.name == partition_column)
-        .unwrap()
-        .stats
-        .as_ref()
-        .unwrap()
-    {
+fn get_file_bounds(
+    file: &manifest::File,
+    partition_column: String,
+) -> (DateTime<Utc>, DateTime<Utc>) {
+    let col = match file.columns().iter().find(|col| col.name == partition_column) {
+        Some(c) => c,
+        None => {
+            error!("Partition column {partition_column} not found in manifest file {}", file.file_path);
+            // Fallback: treat as empty range to avoid crash; adjust if a different strategy is preferred
+            return (DateTime::from_timestamp_millis(0).unwrap(), DateTime::from_timestamp_millis(0).unwrap());
+        }
+    };
+    let stats = match col.stats.as_ref() {
+        Some(s) => s,
+        None => {
+            error!("Missing stats for column {partition_column} in {}", file.file_path);
+            return (DateTime::from_timestamp_millis(0).unwrap(), DateTime::from_timestamp_millis(0).unwrap());
+        }
+    };
+    match stats {
         column::TypedStatistics::Int(stats) => (
             DateTime::from_timestamp_millis(stats.min).unwrap(),
             DateTime::from_timestamp_millis(stats.max).unwrap(),
         ),
         _ => unreachable!(),
     }
 }

If a hard error is preferred instead of a fallback, we can return a Result and propagate it.

src/utils/json/mod.rs (1)

195-217: Document wrapping of array inputs in convert_array_to_object

Call sites and tests confirm that when body is a JSON array (and no partitioning), the entire array is returned as a single Vec<Value>. Please update the docstring on src/utils/json/mod.rs (around the convert_array_to_object signature, ~line 219) to explicitly state:

• For array inputs without partitioning, the function wraps the full array into a one-element Vec.

[src/utils/json/mod.rs:219]

src/parseable/streams.rs (1)

1368-1371: Avoid println in tests; assert update is correct.

The updated assertion to expect 3 Arrow files is correct for the same-minute case before compaction. Replace the println with tracing to keep tests clean.

Apply this diff:

-        println!("arrow files: {:?}", staging.arrow_files());
+        tracing::debug!("arrow files: {:?}", staging.arrow_files());

src/parseable/mod.rs (1)

922-931: Custom partition validation limits to a single key.

Matches the documented constraint and simplifies downstream layout. Consider updating the error message to include the provided value for quicker debugging.

Apply this diff:

-        return Err(CreateStreamError::Custom {
-            msg: "Maximum 1 custom partition key is supported".to_string(),
+        return Err(CreateStreamError::Custom {
+            msg: format!("Maximum 1 custom partition key is supported (got: {custom_partition})"),
             status: StatusCode::BAD_REQUEST,
         });

src/storage/field_stats.rs (3)

180-181: Lowering log level to trace may hide actionable failures

These failures previously surfaced at warn!. Moving to trace! will make them easy to miss in production. Consider debug! with field context (stream/field) so operators can correlate silent drops.

- trace!("Failed to execute distinct stats query: {e}");
+ debug!("Failed to execute distinct stats query: {e}");

---

188-189: Same concern: batch fetch errors downgraded to trace

If the stream is noisy, this will fly under the radar. Suggest debug! and include stream_name/field_name to aid troubleshooting.

- trace!("Failed to fetch batch in distinct stats query: {e}");
+ debug!("Failed to fetch batch in distinct stats query: {e}");

---

216-217: Field-specific query failure should not be fully silent

trace! here likely hides intermittent schema/data issues. Recommend debug! (or keep warn!) to retain visibility.

- trace!("Failed to execute distinct stats query for field: {field_name}, error: {e}");
+ debug!("Failed to execute distinct stats query for field: {field_name}, error: {e}");

src/handlers/http/ingest.rs (1)

515-545: HTTP status for CustomError

PostError::CustomError maps to 500. For client-side misuse (e.g., bad header combo), 400 is more appropriate. Consider a dedicated variant (e.g., TimePartitionIngestionError) mapped to 400 instead of reusing CustomError.

src/catalog/mod.rs (1)

130-158: Parallel grouping is fine; consider determinism only if needed

Using Rayon to group by partition is fine. If deterministic processing order ever matters, add a final sort of keys before iteration. Not required right now.

src/storage/object_storage.rs (2)

972-999: Minor: spawn_parquet_upload_task doesn’t need to be async

This function only schedules a task; making it fn avoids an unnecessary .await.

-async fn spawn_parquet_upload_task(
+fn spawn_parquet_upload_task(
     join_set: &mut JoinSet<Result<UploadResult, ObjectStorageError>>,
@@
-) {
+) {
@@
-    join_set.spawn(async move {
+    join_set.spawn(async move {
         let _permit = semaphore.acquire().await.expect("semaphore is not closed");
 
         upload_single_parquet_file(store, path, stream_relative_path, stream_name, schema).await
     });
-}
+}

---

1001-1044: Result collection strategy is fine; consider best-effort behavior on single-file errors

Currently, the first per-file error aborts the entire sync with an error, potentially leaving other tasks running until JoinSet is dropped. If you prefer best-effort uploads, collect all successes and log failures, then proceed to snapshot update. Optional.

src/utils/json/mod.rs (5)

66-75: De-duplicate gating logic: reuse this helper in flatten_json_body for consistency

You’ve introduced should_apply_generic_flattening() but flatten_json_body re-implements the same condition inline. Using the helper there avoids drift and keeps behavior consistent across partitioned and non-partitioned paths.

Outside this hunk, consider updating flatten_json_body as:

let mut nested_value = if should_apply_generic_flattening(&body, schema_version, log_source) {
    let flattened_json = generic_flattening(&body)?;
    convert_to_array(flattened_json)?
} else {
    body
};

---

77-115: Simplify: always iterate the flattened results (remove special-casing len == 1)

Both branches call flatten::flatten on each resulting item; the only difference is avoiding a tiny loop for len == 1. Unify to reduce code and potential divergence.

Apply this diff:

 fn apply_generic_flattening_for_partition(
     element: Value,
     time_partition: Option<&String>,
     time_partition_limit: Option<NonZeroU32>,
     custom_partition: Option<&String>,
 ) -> Result<Vec<Value>, anyhow::Error> {
-    let flattened_json = generic_flattening(&element)?;
-
-    if flattened_json.len() == 1 {
-        // Single result - process normally
-        let mut nested_value = flattened_json.into_iter().next().unwrap();
-        flatten::flatten(
-            &mut nested_value,
-            "_",
-            time_partition,
-            time_partition_limit,
-            custom_partition,
-            true,
-        )?;
-        Ok(vec![nested_value])
-    } else {
-        // Multiple results - process each individually
-        let mut result = Vec::new();
-        for item in flattened_json {
-            let mut processed_item = item;
-            flatten::flatten(
-                &mut processed_item,
-                "_",
-                time_partition,
-                time_partition_limit,
-                custom_partition,
-                true,
-            )?;
-            result.push(processed_item);
-        }
-        Ok(result)
-    }
+    let flattened_json = generic_flattening(&element)?;
+    let mut result = Vec::with_capacity(flattened_json.len());
+    for mut item in flattened_json {
+        flatten::flatten(
+            &mut item,
+            "_",
+            time_partition,
+            time_partition_limit,
+            custom_partition,
+            true,
+        )?;
+        result.push(item);
+    }
+    Ok(result)
 }

---

149-171: Pre-allocate results; consider parallelizing for large batches

The loop is fine. Two small options:

• Pre-allocate: result.reserve(arr.len()) as a lower bound.
• Optional future: use rayon to parallelize per-element processing if CPU-bound and safe (PR mentions rayon was added elsewhere).

Minimal pre-allocation diff:

 fn process_partitioned_array(
@@
 ) -> Result<Vec<Value>, anyhow::Error> {
-    let mut result = Vec::new();
+    let mut result = Vec::new();
+    result.reserve(arr.len()); // lower bound; items can expand

If/when you choose to parallelize, you can adopt rayon and then flatten:

// Outside this hunk, add once: use rayon::prelude::*;
let chunks: Result<Vec<Vec<Value>>, _> = arr
    .into_par_iter()
    .map(|element| {
        process_partitioned_element(
            element,
            time_partition,
            time_partition_limit,
            custom_partition,
            schema_version,
            log_source,
        )
    })
    .collect();

let result = chunks?.into_iter().flatten().collect();

---

483-515: Add a negative test: missing time-partition field should reject the whole batch

Given the PR objective (reject batch if any event lacks the partition field), add a test where one element omits source_time and assert that convert_array_to_object returns Err.

Add alongside this test:

#[test]
fn test_convert_array_to_object_with_time_partition_missing_field_rejects_batch() {
    let json = json!([
        { "a": "b", "source_time": "2025-08-01T00:00:00.000Z" },
        { "a": "b" } // missing required partition field
    ]);

    let time_partition = Some("source_time".to_string());
    let result = convert_array_to_object(
        json,
        time_partition.as_ref(),
        None,
        None,
        SchemaVersion::V0,
        &crate::event::format::LogSource::default(),
    );

    assert!(result.is_err(), "Batch should be rejected if any event lacks time partition field");
}

---

517-545: Consider adding a SchemaVersion::V1 case to exercise generic flattening gating

This test validates the new non-partitioned batching, but only for V0. A small V1 case with shallow nesting would exercise should_apply_generic_flattening and guard against regressions in the gating condition.

Example to add near this test:

#[test]
fn test_non_partitioned_v1_generic_flattening_applies() {
    // Single object with an array to be expanded by generic_flattening
    let json = json!({
        "id": 1,
        "items": [ {"x": 10}, {"x": 20} ]
    });

    let result = convert_array_to_object(
        json.clone(),
        None,
        None,
        None,
        SchemaVersion::V1,
        &crate::event::format::LogSource::Json, // matches gating
    );

    assert!(result.is_ok());
    let out = result.unwrap();
    assert_eq!(out.len(), 1, "Non-partitioned path should still return a single batch item");
    // And the single batch item should be an array with expanded objects
    // depending on flatten_json_body's V1 behavior.
    // You can assert shape here based on expected expansion semantics.
}

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1263
File: src/handlers/http/ingest.rs:300-310
Timestamp: 2025-03-26T06:44:53.362Z
Learning: In Parseable, every stream is always associated with a log_source - no stream can exist without a log_source. For otel-traces and otel-metrics, strict restrictions are implemented where ingestion is rejected if a stream already has a different log_source format. However, regular logs from multiple log_sources can coexist in a single stream.

Learnt from: de-sh
PR: parseablehq/parseable#1185
File: src/handlers/http/logstream.rs:255-261
Timestamp: 2025-02-14T09:49:25.818Z
Learning: In Parseable's logstream handlers, stream existence checks must be performed for both query and standalone modes. The pattern `!PARSEABLE.streams.contains(&stream_name) && (PARSEABLE.options.mode != Mode::Query || !PARSEABLE.create_stream_and_schema_from_storage(&stream_name).await?)` ensures proper error handling in both modes.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/parseable/streams.rs:319-331
Timestamp: 2025-06-16T09:50:38.636Z
Learning: In Parseable's Ingest or Query mode, the node_id is always available because it's generated during server initialization itself, before the get_node_id_string() function in streams.rs would be called. This makes the .expect() calls on QUERIER_META.get() and INGESTOR_META.get() safe in this context.